package com.bfxy.disruptor.common;

import java.util.concurrent.atomic.AtomicBoolean;

public final class BatchEventProcessor<T> implements EventProcessor {//继承Runnable，所以在线程中运行。

	//判断是否运行着
    private final AtomicBoolean running = new AtomicBoolean(false);
    //运行过程中遇到异常就处理
    private ExceptionHandler<? super T> exceptionHandler = new FatalExceptionHandler();
    private final DataProvider<T> dataProvider;//通过序号获取数据，就是RingBuffer。
    //ProcessingSequenceBarrier implements SequenceBarrier
    private final SequenceBarrier sequenceBarrier;//SequenceBarrier栅栏：生产与消费进度协调的类，
    private final EventHandler<? super T> eventHandler;//消费者 
    private final Sequence sequence = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);//ringbuffer里面最大的序号。
    private final TimeoutHandler timeoutHandler;//超时时候做的处理

    public BatchEventProcessor(final DataProvider<T> dataProvider, final SequenceBarrier sequenceBarrier,
                               final EventHandler<? super T> eventHandler) {
        this.dataProvider = dataProvider;
        this.sequenceBarrier = sequenceBarrier;
        this.eventHandler = eventHandler;

        if (eventHandler instanceof SequenceReportingEventHandler) {
            ((SequenceReportingEventHandler<?>)eventHandler).setSequenceCallback(sequence);
        }

        timeoutHandler = (eventHandler instanceof TimeoutHandler) ? (TimeoutHandler) eventHandler : null;
    }

    @Override
    public Sequence getSequence()  {
        return sequence;
    }

    @Override
    public void halt() {
        running.set(false);
        sequenceBarrier.alert();
    }

    @Override
    public boolean isRunning() {
        return running.get();
    }

    public void setExceptionHandler(final ExceptionHandler<? super T> exceptionHandler) {
        if (null == exceptionHandler) {
            throw new NullPointerException();
        }

        this.exceptionHandler = exceptionHandler;
    }

    /**
     * It is ok to have another thread rerun this method after a halt().
     */
    @Override
    public void run() {
        if (!running.compareAndSet(false, true))//CAS操作，running是false就置为true，running是true就抛出异常
        {
            throw new IllegalStateException("Thread is already running");
        }
        sequenceBarrier.clearAlert();//序号栅栏清空

        notifyStart();//唤醒线程去工作

        T event = null;//实际的对象
        
        //下一个消费者想要的序号
        long nextSequence = sequence.get() + 1L;//下一个可用序号。sequence = ringbuffer里面最大的序号。
        try {//2层while循环
            while (true) {//死循环，一直等待下一个序号。
                try {
                	//nextSequence是下一个消费者想要的序号，availableSequence是真实的序号
                	//消费者消费时候，生产者在生产，就会导致序号不一致。
                    final long availableSequence = sequenceBarrier.waitFor(nextSequence);//下次等待14
                    //如果消费者想要10，但是生产了很多availableSequence=13，一直去消费，直到消费完。
                    while (nextSequence <= availableSequence) {
                        event = dataProvider.get(nextSequence);//RingBuffer取到真实的数据
                        //消费者消费
                        eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);
                        nextSequence++;//14
                    }

                    sequence.set(availableSequence);//13
                }
                //抛出异常时候，死循环while退出。
                catch (final TimeoutException e) {//抛出超时异常时while退出
                    notifyTimeout(sequence.get());
                }
                catch (final AlertException ex) {//抛出AlertException异常while就退出，捕获到异常就停止return
                    if (!running.get()) {
                        break;
                    }
                }
                catch (final Throwable ex) {//抛出异常while退出
                    exceptionHandler.handleEventException(ex, nextSequence, event);
                    sequence.set(nextSequence);
                    nextSequence++;
                }
            }
        }
        finally {
            notifyShutdown();
            running.set(false);
        }
    }

    private void notifyTimeout(final long availableSequence) {
        try {
            if (timeoutHandler != null) {
                timeoutHandler.onTimeout(availableSequence);
            }
        }
        catch (Throwable e) {
            exceptionHandler.handleEventException(e, availableSequence, null);
        }
    }

    /**
     * Notifies the EventHandler when this processor is starting up
     */
    private void notifyStart() {
        if (eventHandler instanceof LifecycleAware) {
            try {
                ((LifecycleAware)eventHandler).onStart();
            }
            catch (final Throwable ex) {
                exceptionHandler.handleOnStartException(ex);
            }
        }
    }

    /**
     * Notifies the EventHandler immediately prior to this processor shutting down
     */
    private void notifyShutdown() {
        if (eventHandler instanceof LifecycleAware) {
            try {
                ((LifecycleAware)eventHandler).onShutdown();
            }
            catch (final Throwable ex) {
                exceptionHandler.handleOnShutdownException(ex);
            }
        }
    }
}